package com.hucais.sync.hive2es

import cn.hutool.core.date.DateUtil
import com.hucais.core.utils.{CommonUtils, DefaultPropertiesUtil, SparkDefaultUtil}
import org.elasticsearch.spark.sql.EsSparkSQL

object SyncOdsRanking {
  def main(args: Array[String]): Unit = {
    val sparkSession = SparkDefaultUtil.getSession(isLocal = true,this.getClass.getSimpleName,withEs = true,withHive = true)
    val ssc = sparkSession.sparkContext
    ssc.hadoopConfiguration.set("fs.defaultFS", DefaultPropertiesUtil.get("fs.defaultFS"))

    //获取ODS层的数据
    import sparkSession.implicits._
    val tmpDF = sparkSession.sql(
      """
        |select channel,year,ranking_type,ranking_name,detail
        |from
        |(
        |select channel,year,ranking_type,ranking_name,detail,
        |row_number() over(partition by channel,year,ranking_type,ranking_name order by acquisition_time desc) as rn
        |from published.ods_ranking
        |)a where a.rn=1
        |""".stripMargin)

    val resultDS = tmpDF.mapPartitions(partitions => {
      partitions
        .filter(item => CommonUtils.isNotBlankExt(item.getString(0)) && CommonUtils.isNotBlankExt(item.getString(1)) &&
          CommonUtils.isNotBlankExt(item.getString(2)) && CommonUtils.isNotBlankExt(item.getString(3)))
        .map(item => {
          val channel = CommonUtils.filterSomeStr(item.getString(0))
          val year = CommonUtils.filterSomeStr(item.getString(1))
          val ranking_type = CommonUtils.filterSomeStr(item.getString(2))
          val ranking_name = CommonUtils.filterSomeStr(item.getString(3))

          val currentTime = DateUtil.format(new java.util.Date(), "yyyy-MM-dd HH:mm:ssZ").replace(" ", "T")
          TmpRanking(channel, year, ranking_type, ranking_name, currentTime,item.getString(4))
        })
    })

    EsSparkSQL.saveToEs(resultDS, DefaultPropertiesUtil.get("ods.ranking"))
  }

  case class TmpRanking(
                         channel: String,
                         year: String,
                         ranking_type: String,
                         ranking_name: String,
                         create_time : String,
                         detail: String
                       )


}
